package com.ams.server.handler;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Random;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ams.io.BufferFactory;
import com.ams.io.network.connection.Connection;
import com.ams.protocol.http.HTTP;
import com.ams.protocol.http.HttpRequest;
import com.ams.protocol.http.HttpResponse;
import com.ams.protocol.http.MimeTypes;
import com.ams.util.BufferUtils;

public class RtmpTunnelHandler extends ProtocolHandler {
    final private Logger logger = LoggerFactory
            .getLogger(RtmpTunnelHandler.class);

    private String docRoot;
    private static HashMap<String, Connection> sessions = new HashMap<String, Connection>();
    private static byte[] pollInterval = { (byte) 0x01, (byte) 0x03,
            (byte) 0x05, (byte) 0x09, (byte) 0x11, (byte) 0x21 };
    private static int currentInterval = 0;
    private static int emptyReplies = 0;

    public RtmpTunnelHandler(String docRoot) {
        this.docRoot = docRoot;
    }

    private String newSessionId() {
        byte[] seed = new byte[15];
        Random rnd = new Random();
        rnd.nextBytes(seed);
        StringBuffer hexString = new StringBuffer();
        for (int i = 0; i < seed.length; i++) {
            hexString.append(Integer.toHexString(0xFF & seed[i]));
        }

        return hexString.toString();
    }

    private String getSessionId(String loc) {
        return loc.split("/")[2];
    }

    private void nextInterval() {
        if (currentInterval < pollInterval.length - 1) {
            currentInterval++;
        }
    }

    private byte getInterval(boolean isEmpty) {
        if (isEmpty) {
            emptyReplies++;
            if (emptyReplies >= 10) {
                emptyReplies = 0;
                nextInterval();
            }
        } else {
            currentInterval = 0;
        }
        return pollInterval[currentInterval];
    }

    private Connection launchHandler(ProtocolHandler handler) {
        Connection conn = new Connection();
        conn.open();
        handler.setConnection(conn);
        // put to worker queue to run
        executor.execute(handler);
        return conn;
    }

    private void openCommand(HttpRequest request, HttpResponse response)
            throws IOException {
        String sessionId = newSessionId();

        // create rtmp handle
        Connection tunnelConn = launchHandler(new RtmpHandler(docRoot));

        sessions.put(sessionId, tunnelConn);

        // return session id
        response.print(sessionId);
        response.print("\n");
        response.flush();
    }

    private void sendCommand(HttpRequest request, HttpResponse response)
            throws IOException {
        String sessionId = getSessionId(request.getLocation());
        if (sessionId == null) {
            response.flush();
            return;
        }
        Connection tunnelConn = sessions.get(sessionId);
        if (tunnelConn == null) {
            response.flush();
            return;
        }
        ByteBuffer[] buf = request.getRawPostBodyData();
        tunnelConn.offerInBuffers(buf);
        ByteBuffer[] data = tunnelConn.pollOutBuffers();

        // return interval & RTMP data
        responseData(response, data);
    }

    private void idleCommand(HttpRequest request, HttpResponse response)
            throws IOException {
        String sessionId = getSessionId(request.getLocation());
        if (sessionId == null) {
            response.flush();
            return;
        }
        Connection tunnelConn = sessions.get(sessionId);
        if (tunnelConn == null) {
            response.flush();
            return;
        }
        ByteBuffer[] data = tunnelConn.pollOutBuffers();
        // return interval & RTMP data
        responseData(response, data);
    }

    private void closeCommand(HttpRequest request, HttpResponse response)
            throws IOException {
        String sessionId = getSessionId(request.getLocation());
        if (sessionId == null) {
            response.flush();
            return;
        }
        Connection tunnelConn = sessions.get(sessionId);
        sessions.remove(sessionId);
        if (tunnelConn != null) {
            tunnelConn.close();
        }
        // replay 0x00
        ByteBuffer[] buf = new ByteBuffer[1];
        buf[0] = BufferFactory.allocate(1);
        buf[0].put((byte) 0x00);
        buf[0].flip();
        response.flush(buf);
    }

    private void responseData(HttpResponse response, ByteBuffer[] data)
            throws IOException {
        ByteBuffer[] buf = new ByteBuffer[1];
        buf[0] = BufferFactory.allocate(1);
        byte interval = getInterval(data.length == 0);
        buf[0].put(interval);
        buf[0].flip();
        response.flush(BufferUtils.concat(buf, data));
    }

    private void responseError(HttpResponse response, int code)
            throws IOException {
        response.setContentType(MimeTypes.getContentType("text"));
        response.setHttpResult(code);
        response.flush();
    }

    public void run() {
        HttpRequest request = new HttpRequest(conn.getInputStream());
        HttpResponse response = new HttpResponse(conn.getOutputStream());
        try {
            request.parse();

            response.setContentType(MimeTypes.getContentType("fcs"));
            response.setKeepAlive(true);
            response.setHttpResult(HTTP.HTTP_OK);

            if (HTTP.HTTP_METHOD_POST == request.getMethod()) {
                // OPEN, SEND, IDLE, CLOSE command
                String loc = request.getLocation().toUpperCase();
                if (loc.startsWith("/OPEN")) {
                    openCommand(request, response);
                } else if (loc.startsWith("/SEND")) {
                    sendCommand(request, response);
                } else if (loc.startsWith("/IDLE")) {
                    idleCommand(request, response);
                } else if (loc.startsWith("/CLOSE")) {
                    closeCommand(request, response);
                } else {
                    responseError(response, HTTP.HTTP_BAD_REQUEST);
                }
            }
            if (request.isKeepAlive()) {
                conn.close(true);
            } else {
                conn.close();
            }
        } catch (IOException e) {
            logger.debug(e.getMessage());
            try {
                responseError(response, HTTP.HTTP_INTERNAL_ERROR);
            } catch (IOException e1) {
            }
            close();
        }
    }

    public ProtocolHandler handlerFactory() {
        return new RtmpTunnelHandler(docRoot);
    }

}